package com.runa.monitor.platinterface.mqtt;

import com.runa.monitor.platinterface.extend.hdb.CollectionUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 2024/7/24 15:27
 * zhangYB
 */
@Slf4j
public class MqttAsyncClientPool {


    private String clientId;

    private MQTTExecutor mqttExecutor;

    private List<MQTTAsyncClient> consumerClients = new ArrayList<>();

    private List<MQTTAsyncClient> publisherClients = new ArrayList<>();

    private AtomicInteger consumerIndex = new AtomicInteger(0);

    private AtomicInteger publisherIndex = new AtomicInteger(0);


    public MqttAsyncClientPool(String clientId, MQTTExecutor mqttExecutor) {
        this.clientId = clientId;
        this.mqttExecutor = mqttExecutor;
        createClient();
    }


    private void createClient() {
        new Thread(() -> {
            for (Integer i = 0; i < EmqConfig.getPublisherClients(); i++) {
                MQTTAsyncClient mqttAsyncClient = createMQTTAsyncClient("P", publisherClients.size());
                if (null != mqttAsyncClient) {
                    publisherClients.add(mqttAsyncClient);
                }
            }
            for (Integer i = 0; i < EmqConfig.getConsumerClients(); i++) {
                MQTTAsyncClient mqttAsyncClient = createMQTTAsyncClient("C", consumerClients.size());
                if (null != mqttAsyncClient) {
                    consumerClients.add(mqttAsyncClient);
                }
            }
        }, "MQTT-Async-Client-Creator").start();
//        try {
//            TimeUnit.SECONDS.sleep(10);
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
    }

    private MQTTAsyncClient createMQTTAsyncClient(String type, int index) {
        try {
            MQTTAsyncClient client = new MQTTAsyncClient(EmqConfig.getHOST(), clientId, type, index, new MemoryPersistence(), mqttExecutor, this);
            return client;
        } catch (Exception e) {
            log.error("", e);
        }
        return null;
    }

    public void create() {

    }


    public MQTTAsyncClient getConsumer(Integer index) {
        return consumerClients.get(index);
    }

    public MQTTAsyncClient nextConsumer() {
        return next(consumerClients, consumerIndex);
    }

    public MQTTAsyncClient getPublisher(Integer index) {
        return publisherClients.get(index);
    }

    public MQTTAsyncClient nextPublisher() {
        return next(publisherClients, publisherIndex);
    }

    private MQTTAsyncClient next(List<MQTTAsyncClient> clients, AtomicInteger index) {
        if (CollectionUtil.isEmpty(clients)) {
            return null;
        }
        Integer nextIndex = nextIndex(index, clients.size());
        return clients.get(nextIndex);
    }

    private Integer nextIndex(AtomicInteger counter, Integer totalCount) {
        if (totalCount % 2 == 0) {
            return counter.getAndIncrement() & totalCount - 1;
        } else {
            return counter.getAndIncrement() % totalCount;
        }
    }





}
